#include "utilities.h"

//This is the read thread which will receive the request from other nodes and pass it on to the dispatcher.
void *readfun(void *ptr)
{
	struct timeval readtime;
	int numbytes = -1;
	writeStruct wrStruct = *((writeStruct*) ptr);
	if(printFlag==1)
	{
		printf("Read Thread for Portnumber %d Starts\n",wrStruct.portNumber);
	}
	while(!dieChild)
	{
		allTimers tempTimer;
		char headBuff[27];
		if((numbytes = read(wrStruct.sock, headBuff, 27)) == 0)
		{
			pthread_mutex_lock(&notifyLock);
			notifyCount--;
			pthread_mutex_unlock(&notifyLock);
			//Now clear the map
			pthread_mutex_lock(&writeMapMutex);
			if(writeMap.find(wrStruct.portNumber) != writeMap.end())
			{
				writeMap.erase(writeMap.find(wrStruct.portNumber));
			}
			if(printFlag==1)
			{
				cout<<"eSIZE IN READ "<<writeMap.size()<<endl;
			}
			pthread_mutex_unlock(&writeMapMutex);
			pthread_mutex_lock(&timerlock);
			myTimers.erase(myTimers.find(wrStruct.portNumber));
			pthread_mutex_unlock(&timerlock);
			pthread_mutex_lock(&maplock);
			myMap.erase(myMap.find(wrStruct.portNumber));
			pthread_mutex_unlock(&maplock);
			//Now clean shutdown the system
			pthread_mutex_lock(&notifyLock);
			if(notifyCount == 0)
			{
				serverShutdown = true;
				dieChild = true;
				if(printFlag==1)
				{
				cout<<statusFile<<endl;
				}
				close(globalAcceptSock);
			}
			pthread_mutex_unlock(&notifyLock);
			if(printFlag==1)
			{
			cout<<"Error while reading"<<endl;
			}
			close(wrStruct.sock);
			pthread_exit(NULL);
		}
		if(numbytes == -1)
		{
			//Now clear the map
			pthread_mutex_lock(&writeMapMutex);
			if(writeMap.find(wrStruct.portNumber) != writeMap.end())
			{
				writeMap.erase(writeMap.find(wrStruct.portNumber));
			}
			if(printFlag==1)
			{
				cout<<"SIZE IN READ "<<writeMap.size()<<endl;
			}
			pthread_mutex_unlock(&writeMapMutex);
			pthread_mutex_lock(&timerlock);
			myTimers.erase(myTimers.find(wrStruct.portNumber));
			pthread_mutex_unlock(&timerlock);
			pthread_mutex_lock(&maplock);
			myMap.erase(myMap.find(wrStruct.portNumber));
			pthread_mutex_unlock(&maplock);
			if(printFlag==1)
			{
				cout<<"Owned socket is closed - "<<wrStruct.sock<<endl;
			}
			close(wrStruct.sock);
			pthread_exit(NULL);
		}
		if(dieChild)
		{
			close(wrStruct.sock);
			pthread_exit(NULL);
		}
		pthread_mutex_lock(&timerlock);
		tempTimer = myTimers[wrStruct.portNumber];
		tempTimer.keepAliveTime = 0;
		myTimers[wrStruct.portNumber] = tempTimer;
		pthread_mutex_unlock(&timerlock);
		msg_header temp;
		fillMsgHdrStructure(temp, headBuff);
		char dataBuff[temp.datalength];
		if(temp.datalength != 0)
		{
			if((numbytes = read(wrStruct.sock, dataBuff, temp.datalength)) == 0)
			{
				pthread_mutex_lock(&notifyLock);
				notifyCount--;
				pthread_mutex_unlock(&notifyLock);
				//Now clear the map
				pthread_mutex_lock(&writeMapMutex);
				if(writeMap.find(wrStruct.portNumber) != writeMap.end())
				{
				writeMap.erase(writeMap.find(wrStruct.portNumber));
				}
				pthread_mutex_unlock(&writeMapMutex);
				pthread_mutex_lock(&timerlock);
				myTimers.erase(myTimers.find(wrStruct.portNumber));
				pthread_mutex_unlock(&timerlock);
				pthread_mutex_lock(&maplock);
				myMap.erase(myMap.find(wrStruct.portNumber));
				pthread_mutex_unlock(&maplock);
				//Now clean shutdown the system
				pthread_mutex_lock(&notifyLock);
				if(notifyCount == 0)
				{
					
					serverShutdown = true;
					dieChild = true;
					if(printFlag==1)
					{
						cout<<statusFile<<endl;
					}
					close(globalAcceptSock);
				}
				pthread_mutex_unlock(&notifyLock);
				if(printFlag==1)
				{
					cout<<"Error while reading"<<endl;
				}
				close(wrStruct.sock);
				pthread_exit(NULL);
			}
			if(numbytes == -1)
			{
				//Now clear the map
				pthread_mutex_lock(&writeMapMutex);
				if(writeMap.find(wrStruct.portNumber) != writeMap.end())
				{
					writeMap.erase(writeMap.find(wrStruct.portNumber));
				}
				pthread_mutex_unlock(&writeMapMutex);
				pthread_mutex_lock(&timerlock);
				myTimers.erase(myTimers.find(wrStruct.portNumber));
				pthread_mutex_unlock(&timerlock);
				pthread_mutex_lock(&maplock);
				myMap.erase(myMap.find(wrStruct.portNumber));
				pthread_mutex_unlock(&maplock);
				if(printFlag==1)
				{
					cout<<"Owned socket is closed"<<endl;
				}
				close(wrStruct.sock);
				pthread_exit(NULL);
			}
			if(dieChild)
			{
				close(wrStruct.sock);
				pthread_exit(NULL);
			}
		}
		//Do not put in the dispatcher queue if the notify is sent for shutdown
		if(!notifyFlag)
		{
			listBuffStruct lbtemp;
			gettimeofday(&readtime,NULL);
			memset(&lbtemp.listBuff, '\0', 512);
			memcpy(&lbtemp.listBuff, headBuff, sizeof(headBuff));
			memcpy(&(lbtemp.listBuff[27]), dataBuff, temp.datalength);
			lbtemp.datalength = 27+temp.datalength;
			lbtemp.portNumber = wrStruct.portNumber;
			write_to_log(lbtemp,0,readtime);
			pthread_mutex_lock(&eventQueueMutex);
			eventQueue.push_back(lbtemp);
			pthread_mutex_unlock(&eventQueueMutex);
		}
	}
	close(wrStruct.sock);
	pthread_exit(NULL);
}
